package com.wande.dataplatform.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.core.utils.StringUtils;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.common.satoken.utils.LoginHelper;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.Job;
import com.wande.dataplatform.domain.JobInstance;
import com.wande.dataplatform.domain.bo.JobBo;
import com.wande.dataplatform.domain.vo.JobInstanceVO;
import com.wande.dataplatform.domain.vo.JobVO;
import com.wande.dataplatform.mapper.JobInstanceMapper;
import com.wande.dataplatform.mapper.JobMapper;
import com.wande.dataplatform.service.IDolphinSchedulerService;
import com.wande.dataplatform.service.IJobService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;

/**
 * 作业管理Service实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class JobServiceImpl implements IJobService {

    private final JobMapper jobMapper;
    private final JobInstanceMapper jobInstanceMapper;
    private final IDolphinSchedulerService dolphinSchedulerService;

    @Override
    public TableDataInfo<JobVO> queryPageList(JobBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<Job> wrapper = buildQueryWrapper(bo);
        Page<JobVO> page = jobMapper.selectVoPage(pageQuery.build(), wrapper);
        return TableDataInfo.build(page);
    }

    @Override
    public List<JobVO> queryList(JobBo bo) {
        LambdaQueryWrapper<Job> wrapper = buildQueryWrapper(bo);
        return jobMapper.selectVoList(wrapper);
    }

    private LambdaQueryWrapper<Job> buildQueryWrapper(JobBo bo) {
        LambdaQueryWrapper<Job> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(Job::getTenantId, LoginHelper.getTenantId());
        wrapper.like(ObjectUtil.isNotEmpty(bo.getName()), Job::getName, bo.getName());
        wrapper.eq(ObjectUtil.isNotEmpty(bo.getJobType()), Job::getJobType, bo.getJobType());
        wrapper.eq(ObjectUtil.isNotEmpty(bo.getStatus()), Job::getStatus, bo.getStatus());
        wrapper.eq(ObjectUtil.isNotEmpty(bo.getProjectId()), Job::getProjectId, bo.getProjectId());
        wrapper.orderByDesc(Job::getCreateTime);
        return wrapper;
    }

    @Override
    public JobVO queryById(Long id) {
        JobVO vo = jobMapper.selectVoById(id);
        if (vo == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_NOT_FOUND.getMessage());
        }
        return vo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long createJob(JobBo bo) {
        Job job = BeanUtil.toBean(bo, Job.class);
        String tenantIdStr = LoginHelper.getTenantId();
        if (StringUtils.isNotBlank(tenantIdStr)) {
            job.setTenantId(Long.parseLong(tenantIdStr));
        }

        // 如果配置了调度，创建DolphinScheduler工作流
        if (ObjectUtil.isNotEmpty(bo.getScheduleType()) && !"MANUAL".equals(bo.getScheduleType())) {
            try {
                Long workflowId = dolphinSchedulerService.createWorkflow(job);
                if (workflowId != null) {
                    job.setDsWorkflowId(workflowId);
                }
            } catch (Exception e) {
                log.error("创建DolphinScheduler工作流失败", e);
                throw new ServiceException("创建调度工作流失败: " + e.getMessage());
            }
        }

        jobMapper.insert(job);
        return job.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateJob(JobBo bo) {
        Job job = jobMapper.selectById(bo.getId());
        if (job == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_NOT_FOUND.getMessage());
        }

        BeanUtil.copyProperties(bo, job);

        // 如果调度配置变更，更新DolphinScheduler工作流
        if (job.getDsWorkflowId() != null) {
            try {
                dolphinSchedulerService.updateWorkflow(job);
            } catch (Exception e) {
                log.error("更新DolphinScheduler工作流失败", e);
                throw new ServiceException("更新调度工作流失败: " + e.getMessage());
            }
        }

        return jobMapper.updateById(job) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteJob(Long id) {
        Job job = jobMapper.selectById(id);
        if (job == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_NOT_FOUND.getMessage());
        }

        // 删除DolphinScheduler工作流
        if (job.getDsWorkflowId() != null) {
            try {
                dolphinSchedulerService.deleteWorkflow(job.getDsWorkflowId());
            } catch (Exception e) {
                log.warn("删除DolphinScheduler工作流失败", e);
            }
        }

        return jobMapper.deleteById(id) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long startJob(Long id, String triggerType) {
        Job job = jobMapper.selectById(id);
        if (job == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_NOT_FOUND.getMessage());
        }

        if (job.getStatus() == 0) {
            throw new ServiceException("作业已禁用，无法启动");
        }

        // 创建作业实例
        JobInstance instance = new JobInstance();
        instance.setJobId(id);
        instance.setStatus("RUNNING");
        instance.setStartTime(LocalDateTime.now());
        instance.setTriggerType(triggerType);
        instance.setRetryCount(0);
        jobInstanceMapper.insert(instance);

        // 启动DolphinScheduler工作流
        try {
            Long dsInstanceId = dolphinSchedulerService.startWorkflow(job.getDsWorkflowId());
            instance.setDsInstanceId(dsInstanceId);
            jobInstanceMapper.updateById(instance);
        } catch (Exception e) {
            log.error("启动DolphinScheduler工作流失败", e);
            instance.setStatus("FAILED");
            instance.setErrorMessage("启动工作流失败: " + e.getMessage());
            instance.setEndTime(LocalDateTime.now());
            jobInstanceMapper.updateById(instance);
            throw new ServiceException("启动作业失败: " + e.getMessage());
        }

        return instance.getId();
    }

    @Override
    public Boolean stopJobInstance(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException("作业实例不存在");
        }

        if (!"RUNNING".equals(instance.getStatus())) {
            throw new ServiceException("作业实例未在运行中");
        }

        // 停止DolphinScheduler工作流
        if (instance.getDsInstanceId() != null) {
            try {
                dolphinSchedulerService.stopWorkflow(instance.getDsInstanceId());
            } catch (Exception e) {
                log.error("停止DolphinScheduler工作流失败", e);
                throw new ServiceException("停止作业失败: " + e.getMessage());
            }
        }

        instance.setStatus("STOPPED");
        instance.setEndTime(LocalDateTime.now());
        if (instance.getStartTime() != null) {
            instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
        }
        return jobInstanceMapper.updateById(instance) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long rerunJobInstance(Long instanceId) {
        JobInstance oldInstance = jobInstanceMapper.selectById(instanceId);
        if (oldInstance == null) {
            throw new ServiceException("作业实例不存在");
        }

        // 创建新实例
        return startJob(oldInstance.getJobId(), "MANUAL");
    }

    @Override
    public TableDataInfo<JobInstanceVO> queryInstanceList(Long jobId, PageQuery pageQuery) {
        Page<JobInstanceVO> page = pageQuery.build();
        List<JobInstanceVO> list = jobInstanceMapper.selectInstanceListWithJobName(jobId);
        
        // 手动分页
        int start = (int) ((page.getCurrent() - 1) * page.getSize());
        int end = Math.min(start + (int) page.getSize(), list.size());
        List<JobInstanceVO> pageList = list.subList(start, end);
        
        page.setRecords(pageList);
        page.setTotal(list.size());
        
        return TableDataInfo.build(page);
    }

    @Override
    public JobInstanceVO queryInstanceById(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException("作业实例不存在");
        }
        
        JobInstanceVO vo = BeanUtil.toBean(instance, JobInstanceVO.class);
        
        // 获取作业名称
        Job job = jobMapper.selectById(instance.getJobId());
        if (job != null) {
            vo.setJobName(job.getName());
        }
        
        return vo;
    }

    @Override
    public void syncInstanceStatus(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null || instance.getDsInstanceId() == null) {
            return;
        }

        try {
            // 从DolphinScheduler同步状态
            String status = dolphinSchedulerService.getWorkflowInstanceStatus(instance.getDsInstanceId());
            instance.setStatus(status);
            
            if ("SUCCESS".equals(status) || "FAILED".equals(status)) {
                instance.setEndTime(LocalDateTime.now());
                if (instance.getStartTime() != null) {
                    instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
                }
            }
            
            jobInstanceMapper.updateById(instance);
        } catch (Exception e) {
            log.error("同步作业实例状态失败", e);
        }
    }

    @Override
    public String getInstanceLogs(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException("作业实例不存在");
        }

        // 如果有DolphinScheduler实例ID，从DS获取日志
        if (instance.getDsInstanceId() != null) {
            try {
                return dolphinSchedulerService.getWorkflowInstanceLogs(instance.getDsInstanceId());
            } catch (Exception e) {
                log.error("获取DolphinScheduler日志失败", e);
            }
        }

        return instance.getLogs() != null ? instance.getLogs() : "";
    }
}
